/*
    * Copyright (C) 2009 The Guava Authors
    *
    * Licensed under the Apache License, Version 2.0 (the "License");
    * you may not use this file except in compliance with the License.
    * You may obtain a copy of the License at
    *
    * http://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.google.common.util.concurrent;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
  
  import com.google.common.annotations.Beta;
  
  import java.util.concurrent.Executor;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.ThreadFactory;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  /**
   * Utilities necessary for working with libraries that supply plain {@link
   * Future} instances. Note that, whenver possible, it is strongly preferred to
   * modify those libraries to return {@code ListenableFuture} directly.
   *
   * @author Sven Mawson
   * @since 10.0 (replacing {@code Futures.makeListenable}, which
   *     existed in 1.0)
   */
  @Beta
  public final class JdkFutureAdapters {
    /**
     * Assigns a thread to the given {@link Future} to provide {@link
     * ListenableFuture} functionality.
     *
     * <p><b>Warning:</b> If the input future does not already implement {@code
     * ListenableFuture}, the returned future will emulate {@link
     * ListenableFuture#addListener} by taking a thread from an internal,
     * unbounded pool at the first call to {@code addListener} and holding it
     * until the future is {@linkplain Future#isDone() done}.
     *
     * <p>Prefer to create {@code ListenableFuture} instances with {@link
     * SettableFuture}, {@link MoreExecutors#listeningDecorator(
     * java.util.concurrent.ExecutorService)}, {@link ListenableFutureTask},
     * {@link AbstractFuture}, and other utilities over creating plain {@code
     * Future} instances to be upgraded to {@code ListenableFuture} after the
     * fact.
     */
    public static <V> ListenableFuture<V> listenInPoolThread(
        Future<V> future) {
      if (future instanceof ListenableFuture) {
        return (ListenableFuture<V>) future;
      }
      return new ListenableFutureAdapter<V>(future);
    }
  
    /**
     * Submits a blocking task for the given {@link Future} to provide {@link
     * ListenableFuture} functionality.
     *
     * <p><b>Warning:</b> If the input future does not already implement {@code
     * ListenableFuture}, the returned future will emulate {@link
     * ListenableFuture#addListener} by submitting a task to the given executor at
     * the first call to {@code addListener}. The task must be started by the
     * executor promptly, or else the returned {@code ListenableFuture} may fail
     * to work.  The task's execution consists of blocking until the input future
     * is {@linkplain Future#isDone() done}, so each call to this method may
     * claim and hold a thread for an arbitrary length of time. Use of bounded
     * executors or other executors that may fail to execute a task promptly may
     * result in deadlocks.
     *
     * <p>Prefer to create {@code ListenableFuture} instances with {@link
     * SettableFuture}, {@link MoreExecutors#listeningDecorator(
     * java.util.concurrent.ExecutorService)}, {@link ListenableFutureTask},
     * {@link AbstractFuture}, and other utilities over creating plain {@code
     * Future} instances to be upgraded to {@code ListenableFuture} after the
     * fact.
     *
     * @since 12.0
     */
    public static <V> ListenableFuture<V> listenInPoolThread(
        Future<V> future, Executor executor) {
      checkNotNull(executor);
      if (future instanceof ListenableFuture) {
        return (ListenableFuture<V>) future;
      }
      return new ListenableFutureAdapter<V>(future, executor);
    }
  
    /**
    * An adapter to turn a {@link Future} into a {@link ListenableFuture}.  This
    * will wait on the future to finish, and when it completes, run the
    * listeners.  This implementation will wait on the source future
    * indefinitely, so if the source future never completes, the adapter will
    * never complete either.
    *
    * <p>If the delegate future is interrupted or throws an unexpected unchecked
    * exception, the listeners will not be invoked.
    */
   private static class ListenableFutureAdapter<V> extends ForwardingFuture<V>
       implements ListenableFuture<V> {
 
     private static final ThreadFactory threadFactory =
         new ThreadFactoryBuilder()
             .setDaemon(true)
             .setNameFormat("ListenableFutureAdapter-thread-%d")
             .build();
     private static final Executor defaultAdapterExecutor =
         Executors.newCachedThreadPool(threadFactory);
 
     private final Executor adapterExecutor;
 
     // The execution list to hold our listeners.
     private final ExecutionList executionList = new ExecutionList();
 
     // This allows us to only start up a thread waiting on the delegate future
     // when the first listener is added.
     private final AtomicBoolean hasListeners = new AtomicBoolean(false);
 
     // The delegate future.
     private final Future<V> delegate;
 
     ListenableFutureAdapter(Future<V> delegate) {
       this(delegate, defaultAdapterExecutor);
     }
 
     ListenableFutureAdapter(Future<V> delegate, Executor adapterExecutor) {
       this.delegate = checkNotNull(delegate);
       this.adapterExecutor = checkNotNull(adapterExecutor);
     }
 
     @Override
     protected Future<V> delegate() {
       return delegate;
     }
 
     @Override
     public void addListener(Runnable listener, Executor exec) {
       executionList.add(listener, exec);
 
       // When a listener is first added, we run a task that will wait for
       // the delegate to finish, and when it is done will run the listeners.
       if (hasListeners.compareAndSet(false, true)) {
         if (delegate.isDone()) {
           // If the delegate is already done, run the execution list
           // immediately on the current thread.
           executionList.execute();
           return;
         }
 
         adapterExecutor.execute(new Runnable() {
           @Override
           public void run() {
             try {
               /*
                * Threads from our private pool are never interrupted. Threads
                * from a user-supplied executor might be, but... what can we do?
                * This is another reason to return a proper ListenableFuture
                * instead of using listenInPoolThread.
                */
               getUninterruptibly(delegate);
             } catch (Error e) {
               throw e;
             } catch (Throwable e) {
               // ExecutionException / CancellationException / RuntimeException
               // The task is done, run the listeners.
             }
             executionList.execute();
           }
         });
       }
     }
   }
 
   private JdkFutureAdapters() {}
 }